package com.rrd.hive.udf.plist;

import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import java.io.*;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.rrd.hive.udf.plist.utils.PlistCommon.isExits;
import static com.rrd.hive.udf.plist.utils.PlistCommon.loadDataFromFile;

@Description(
        name = "ato_plist_tags",
        value = "_FUNC_(text) - Creates a map by xml string ",
        extended = "parse tags string to map"
)
public class AtoPlistTagsToMap extends UDF {
    private final static String DIM_APP_TAG_02010112 = "/data/dim/app_tag_02010112";
    private final static Map<String,String> app_name_tag = new HashMap<>(300000);
    private static final List<String> tagsList =
            Arrays.asList(
                    "休闲创意", "休闲娱乐", "医疗健康", "图书阅读", "学习教育", "实用工具", "居家生活", "影音视听", "手机游戏", "摄影摄像",
                    "效率办公", "新闻资讯", "旅行交通", "时尚购物", "聊天社交", "赛车体育", "金融理财");
    static {
        try {
            // 初始化 dim.app_tag_02010112 数据
            loadDataInMap(DIM_APP_TAG_02010112);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    private static void loadDataInMap(String filePath) throws IOException {
        final String s = loadDataFromFile(filePath);
        Arrays.stream(s.split("\n", -1)).forEach(lines -> {
            final String[] line = lines.split("\u0001", -1);
            if (line.length >= 3) {
                final String app_name = line[0].toLowerCase().trim();
                final String apk_name = line[1].toLowerCase();
                final String tag = line[2];
                isExits(app_name_tag, apk_name, tag);
                isExits(app_name_tag, app_name, tag);
            }
        });
    }
    public Map<String,String> evaluate(String app_name,String apk_name) throws HiveException, InterruptedException {
        final Map<String, String> result = new HashMap<>(1);
        StringBuilder stringBuilder = new StringBuilder();
        final boolean isHFQ = "好分期".equals(app_name);
        // hfq_tag
        addFlag(stringBuilder,isHFQ);
        // loan_tag
        if (!isHFQ) {
            final boolean i = hiveUdfRegExp(app_name, "借钱|钱包|钱袋|口袋|贷款|贷|借|借贷|借钱|借条|钱|钱庄|下款|普惠|快贷|易贷|金融|呗|分期|融360|闪银|现金|宜人");
            final boolean i1 = hiveUdfRegExp(apk_name,"koudai|jieqian|daikuan|qianbao|loan|qianzhuang|kuaidai|yidai|wallet|fenqi|money|cash|rong360");
            if (i || i1) {
                stringBuilder.append(1);
            } else {
                stringBuilder.append(0);
            }
        } else {
            stringBuilder.append(0);
        }
        // bank_tag
        addFlag(stringBuilder,hiveUdfRegExp(app_name,"银行|信用卡|农商行|银联|bank|掌上生活|渣打财富管理|HAI生活|云南农信|发现精彩|卡动空间|浦大喜奔|工银联e融|无锡农商"));
        final boolean b = hiveUdfRegExp(app_name, "今彩|大馬彩|大马彩|威力彩|六合|乐透|樂透|澳門|星彩|雙贏彩|菠菜|博彩|博金|彩票|投注|Bet|bet|水位|盘口|赌场|娱乐城|银河娱乐|合彩|葡京|澳门|足彩|红单|体彩|外围|开奖|双色球|toto|TOTO|Toto|lottery|Lottery|lotto|Lotto|casino|Casino|老虎机|轮盘|百家乐|Slot|slot|雙色球|老虎機|輪盤|賭場|盤口|百家樂");
        final boolean b1 = hiveUdfRegExp(apk_name, "lottery|caipiao|lotto|casino");
        // bet_tag
        addFlag(stringBuilder, b || b1);
        String tags = null;
        if (apk_name != null) {
            tags = app_name_tag.get(apk_name.toLowerCase());
            if (tags == null && app_name != null) {
                tags = app_name_tag.get(app_name.toLowerCase().trim());
            }
        }
        String finalTags = tags;
        tagsList.forEach(app -> addFlag(stringBuilder, finalTags != null && finalTags.contains(app)));
        result.put("app_tags",stringBuilder.toString());
        return result;
    }
    private void addFlag(StringBuilder stringBuilder, boolean flag) {
        if (flag) {
            stringBuilder.append(1);
        } else {
            stringBuilder.append(0);
        }
    }
    private static boolean hiveUdfRegExp(String input, String pattern) {
        if (input == null) return false;
        Pattern p = Pattern.compile(pattern);
        Matcher matcher = p.matcher(input);
        if (matcher.find(0)) {
            return true;
        }
        return false;
    }
}
